package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.KeywordStats;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.KeywordUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 关键词TOP10统计应用
 * 统计店铺搜索访客数来源最高的10个关键词（仅在"日"和"7天"维度下显示）
 * - KeywordTopNApp.java ：关键词搜索TOP N统计
 * -
 */
/**
 * 流量域关键词搜索TopN应用
 * 负责统计用户搜索关键词的热度排行，分析用户搜索行为
 * 主要功能点：
 * - 从Kafka页面日志主题读取数据
 * - 过滤出搜索行为事件
 * - 提取搜索关键词
 * - 使用Flink水印策略和窗口处理
 * - 统计各个关键词的搜索频次
 * - 计算TopN热门搜索关键词
 * 
 * 数据流向：
 * 页面日志数据 -> KeywordTopNApp -> 处理后输出热门搜索关键词TopN结果
 */
public class KeywordTopNApp {
    private static final Logger log = LoggerFactory.getLogger(KeywordTopNApp.class);
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 读取Kafka的页面日志主题，特别关注搜索页面
        // 直接从ods_traffic主题读取数据，因为BaseLogApp配置从topic_log读取，但测试数据发送到了ods_traffic
        String pageTopic = "ods_traffic";
        String groupId = "keyword_topn_group_v2"; // 使用新的消费者组ID
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId);
        SingleOutputStreamOperator<JSONObject> jsonObjStream = env.addSource(kafkaSource)
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        log.info("接收到Kafka消息: {}", s);
                        return JSON.parseObject(s);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>
                                forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        long ts = System.currentTimeMillis();
                                        try {
                                            if (jsonObject.containsKey("ts")) {
                                                ts = jsonObject.getLong("ts");
                                            }
                                        } catch (Exception e) {
                                            log.warn("时间戳提取失败，使用当前时间: {}", e.getMessage());
                                        }
                                        log.info("提取时间戳: {}", ts);
                                        return ts;
                                    }
                                })
                );

        // 3. 过滤搜索页面并提取关键词
        SingleOutputStreamOperator<KeywordStats> keywordStatsStream = jsonObjStream
                // 首先添加日志记录收到的每条消息
                .map(new MapFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject map(JSONObject jsonObject) throws Exception {
                        log.info("收到页面数据: {}", jsonObject.toJSONString());
                        return jsonObject;
                    }
                })
                .flatMap(new FlatMapFunction<JSONObject, KeywordStats>() {
                    @Override
                    public void flatMap(JSONObject jsonObject, Collector<KeywordStats> out) throws Exception {
                        try {
                            // 检查是否包含page字段
                            if (!jsonObject.containsKey("page")) {
                                log.warn("不包含page字段的数据被过滤: {}", jsonObject.toJSONString());
                                return;
                            }
                            
                            JSONObject page = jsonObject.getJSONObject("page");
                            if (page == null) {
                                log.warn("page字段为空的数据被过滤: {}", jsonObject.toJSONString());
                                return;
                            }
                            
                            // 检查page_id是否为search
                            String pageId = page.getString("page_id");
                            if (!"search".equals(pageId)) {
                                return;
                            }
                            
                            // 检查是否包含item字段
                            if (!page.containsKey("item")) {
                                log.warn("搜索页面但不包含item字段的数据被过滤: {}", jsonObject.toJSONString());
                                return;
                            }
                            
                            String searchKeyword = page.getString("item");
                            if (searchKeyword == null || searchKeyword.isEmpty()) {
                                log.warn("搜索关键词为空的数据被过滤: {}", jsonObject.toJSONString());
                                return;
                            }
                            
                            log.info("通过过滤的搜索页面数据: item={}", searchKeyword);
                            
                            // 确保common和mid字段存在
                            String mid = "unknown";
                            if (jsonObject.containsKey("common")) {
                                JSONObject common = jsonObject.getJSONObject("common");
                                if (common != null && common.containsKey("mid")) {
                                    mid = common.getString("mid");
                                }
                            }
                            
                            // 确保ts字段存在
                            long ts = System.currentTimeMillis();
                            if (jsonObject.containsKey("ts")) {
                                try {
                                    ts = jsonObject.getLong("ts");
                                } catch (Exception e) {
                                    log.warn("时间戳格式错误，使用当前时间: {}", jsonObject.getString("ts"));
                                }
                            }
                            
                            log.info("接收到搜索数据: 用户mid={}, 关键词={}", mid, searchKeyword);
                            
                            // 使用关键词工具进行分词处理
                            List<String> keywords = null;
                            try {
                                keywords = KeywordUtil.splitKeyword(searchKeyword);
                            } catch (Exception e) {
                                log.error("关键词分词失败: {}, 关键词: {}", e.getMessage(), searchKeyword);
                                // 分词失败时，使用原始关键词
                                keywords = new ArrayList<>();
                                keywords.add(searchKeyword);
                            }
                            
                            if (keywords == null || keywords.isEmpty()) {
                                log.warn("分词结果为空: {}", searchKeyword);
                                return;
                            }
                            
                            log.info("分词结果: {}", keywords);
                            
                            // 处理时间戳转换
                            String statsTime;
                            try {
                                statsTime = DateFormatUtil.toDate(ts);
                            } catch (Exception e) {
                                log.warn("时间戳转换失败，使用当前时间: {}", ts);
                                statsTime = DateFormatUtil.toDate(System.currentTimeMillis());
                            }
                            
                            // 初始化统计指标
                            for (String keyword : keywords) {
                                if (keyword == null || keyword.trim().isEmpty()) {
                                    continue;
                                }
                                
                                KeywordStats stats = new KeywordStats();
                                stats.setKeyword(keyword.trim());
                                stats.setStatsTime(statsTime);
                                stats.setStatsType("1d"); // 默认按日统计
                                stats.setSearchCount(1L);
                                stats.setUvCount(1L); // 每个mid算作一个访客
                                stats.setPvCount(1L); // 每个搜索算作一个PV
                                
                                // 从actions字段中提取加购和收藏行为
                                long cartAddCount = 0L;
                                double paymentAmount = 0.0;
                                if (jsonObject.containsKey("actions")) {
                                    try {
                                        for (Object action : jsonObject.getJSONArray("actions")) {
                                            JSONObject actionObj = (JSONObject) action;
                                            if (actionObj != null && actionObj.containsKey("action")) {
                                                String actionType = actionObj.getString("action").toLowerCase();
                                                if ("addcart".equals(actionType)) {
                                                    cartAddCount = 1L;
                                                } else if ("payment".equals(actionType)) {
                                                    // 实际项目中应从订单数据获取支付金额
                                                    paymentAmount = 100.0; // 示例值
                                                }
                                            }
                                        }
                                    } catch (Exception e) {
                                        log.error("解析actions字段异常: {}", e.getMessage());
                                    }
                                }
                                
                                stats.setCartAddCount(cartAddCount);
                                stats.setPaymentAmount(paymentAmount);
                                // 为了兼容ClickHouse的DateTime类型，我们将Date转换为完整的时间格式
                                stats.setCreateTime(new Date()); // 使用当前完整时间
                                
                                log.info("生成关键词统计数据: {}", stats);
                                out.collect(stats);
                            }
                        } catch (Exception e) {
                            log.error("处理搜索数据异常: {}, 数据: {}", e.getMessage(), jsonObject.toJSONString(), e);
                            // 添加应急处理：即使出错，也尝试生成一个简单的统计数据，确保流程能继续
                            try {
                                KeywordStats errorStats = new KeywordStats();
                                errorStats.setKeyword("error_data");
                                errorStats.setStatsTime(DateFormatUtil.toDate(System.currentTimeMillis()));
                                errorStats.setStatsType("1d");
                                errorStats.setSearchCount(1L);
                                errorStats.setUvCount(1L);
                                errorStats.setPvCount(1L);
                                errorStats.setCartAddCount(0L);
                                errorStats.setPaymentAmount(0.0);
                                errorStats.setCreateTime(new Date());
                                out.collect(errorStats);
                            } catch (Exception innerEx) {
                                log.error("应急处理也失败: {}", innerEx.getMessage());
                            }
                        }
                    }
                }).uid("flatMap-search-keyword"); // 添加UID以支持任务重启时的状态恢复

        // 4. 按关键词和统计类型分组，统计指标（测试用1分钟窗口）
        SingleOutputStreamOperator<KeywordStats> aggregatedStream = keywordStatsStream
                .keyBy(new KeySelector<KeywordStats, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(KeywordStats stats) throws Exception {
                        return new Tuple3<>("test_window", 
                                stats.getStatsType(), 
                                stats.getKeyword());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(1))) // 使用1分钟窗口便于测试
                .reduce(new ReduceFunction<KeywordStats>() {
                    @Override
                    public KeywordStats reduce(KeywordStats value1, KeywordStats value2) throws Exception {
                        // 累加各项指标
                        value1.setSearchCount(value1.getSearchCount() + value2.getSearchCount());
                        // UV去重需要更复杂的处理，这里简化处理
                        value1.setUvCount(value1.getUvCount() + value2.getUvCount());
                        value1.setPvCount(value1.getPvCount() + value2.getPvCount());
                        value1.setCartAddCount(value1.getCartAddCount() + value2.getCartAddCount());
                        value1.setPaymentAmount(value1.getPaymentAmount() + value2.getPaymentAmount());
                        return value1;
                    }
                });

        // 5. 添加日志监控聚合结果
        aggregatedStream.map(new MapFunction<KeywordStats, KeywordStats>() {
            @Override
            public KeywordStats map(KeywordStats stats) throws Exception {
                log.info("聚合后的关键词统计数据: {}", stats);
                return stats;
            }
        });

        // 6. 生成简单的TOP N数据（测试用）
        SingleOutputStreamOperator<KeywordStats> topNStream = aggregatedStream
                .map(new MapFunction<KeywordStats, KeywordStats>() {
                    @Override
                    public KeywordStats map(KeywordStats stats) throws Exception {
                        // 为测试数据设置排名
                        stats.setRank(1);
                        log.info("准备写入ClickHouse的数据: {}", stats);
                        return stats;
                    }
                });

        // 7. 打印TOP10关键词以便监控
        topNStream.print("准备写入ClickHouse的关键词数据: ");
        
        // 8. 将结果写入ClickHouse
        // 使用我们改进后的MyClickHouseUtil工具类，它包含更好的错误处理和日期转换逻辑
        topNStream.map(new MapFunction<KeywordStats, KeywordStats>() {
            @Override
            public KeywordStats map(KeywordStats stats) throws Exception {
                log.info("即将使用MyClickHouseUtil写入数据");
                log.info("  statsTime={}", stats.getStatsTime());
                log.info("  statsType={}", stats.getStatsType());
                log.info("  keyword={}", stats.getKeyword());
                log.info("  searchCount={}", stats.getSearchCount());
                log.info("  createTime={} (时间戳: {})", 
                        stats.getCreateTime(), stats.getCreateTime().getTime());
                
                // 确保所有字段都不为null
                if (stats.getStatsTime() == null) stats.setStatsTime(DateFormatUtil.toDate(System.currentTimeMillis()));
                if (stats.getStatsType() == null) stats.setStatsType("1d");
                if (stats.getKeyword() == null) stats.setKeyword("unknown");
                if (stats.getSearchCount() == null) stats.setSearchCount(0L);
                if (stats.getUvCount() == null) stats.setUvCount(0L);
                if (stats.getPvCount() == null) stats.setPvCount(0L);
                if (stats.getCartAddCount() == null) stats.setCartAddCount(0L);
                if (stats.getPaymentAmount() == null) stats.setPaymentAmount(0.0);
                if (stats.getRank() == null) stats.setRank(0);
                if (stats.getCreateTime() == null) stats.setCreateTime(new Date());
                
                log.info("数据验证完成，准备写入ClickHouse");
                return stats;
            }
        }).addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_keyword_topn_window (stats_time, stats_type, keyword, search_count, uv_count, pv_count, cart_add_count, payment_amount, rank, create_time) values(?,?,?,?,?,?,?,?,?,?)"
        )).name("clickhouse-sink").uid("clickhouse-sink-uid");

        // 增加更多调试日志输出
        log.info("ClickHouse Sink 已配置完成，表名: dws_keyword_topn_window");
        log.info("使用改进后的MyClickHouseUtil，它会自动处理Date到DateTime的转换");
        
        // 添加额外的异常处理和监控
        topNStream.addSink(new org.apache.flink.streaming.api.functions.sink.SinkFunction<KeywordStats>() {
            @Override
            public void invoke(KeywordStats value, Context context) throws Exception {
                // 这个sink只用于监控，不实际写入
                System.out.println("[监控] 即将写入ClickHouse的数据: " + value);
                try {
                    // 尝试手动构建并打印插入语句，用于调试
                    String insertSQL = String.format(
                            "INSERT INTO dws_keyword_topn_window VALUES ('%s', '%s', '%s', %d, %d, %d, %d, %f, %d, '%s')",
                            value.getStatsTime(),
                            value.getStatsType(),
                            value.getKeyword(),
                            value.getSearchCount(),
                            value.getUvCount(),
                            value.getPvCount(),
                            value.getCartAddCount(),
                            value.getPaymentAmount(),
                            value.getRank(),
                            new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(value.getCreateTime())
                    );
                    System.out.println("[调试SQL] " + insertSQL);
                } catch (Exception e) {
                    System.err.println("[监控异常] 格式化调试SQL失败: " + e.getMessage());
                }
            }
        }).name("debug-monitor-sink");

        // 9. 执行作业
        env.execute("KeywordTopNApp");
    }
}